package com.bw.gmall.realtime.Day0923;

import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.app.func.DimAsyncFunction;
import com.bw.gmall.realtime.bean.TradeUserSpuOrderBean;
import com.bw.gmall.realtime.utils.DateFormatUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/*
*dbcUtil.queryList     结合redis  做旁路缓存
*
          异步IO
          多线程
          *
* */
public class DwsTradeUserSpuOrderWindow_a_2 {

    public static void main(String[] args) throws Exception {

     //spuid   tmid   a3   a2  a1    订单数   订单综合金额

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        //2.设置并行度
        env.setParallelism(1);

        //3.设置检查点


        //4.加载数据
        DataStreamSource<String> streamSource =
                env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_trade_order_detail", "DwsTradeUserSpuOrderWindow_axxx"));

        //5. 转换成json数据
        SingleOutputStreamOperator<JSONObject> map = streamSource.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                out.collect(JSONObject.parseObject(value));
            }
        });
        //6.保留第一条   保order_deatil_id   order_info_id
        SingleOutputStreamOperator<JSONObject> jsonObjectSingleOutputStreamOperator = map.keyBy(a -> a.getString("id"))
                .process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    ValueState<String> valueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("valueState", String.class);
                        valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(5l))
                                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).
                                        build());
                        valueState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject value, Context ctx, Collector<JSONObject> out) throws Exception {
                        String v = valueState.value();

                        if (v == null) {
                            out.collect(value);
                            valueState.update("a");
                        }


                    }
                }).assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(10l))
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                return DateFormatUtil.toTs(element.getString("create_time"), true);
                            }
                        }));

        //7.keyby   spuid   tmid   a3   a2  a1    订单数   订单综合金额
        SingleOutputStreamOperator<TradeUserSpuOrderBean> wTMap = jsonObjectSingleOutputStreamOperator.map(new MapFunction<JSONObject, TradeUserSpuOrderBean>() {
            @Override
            public TradeUserSpuOrderBean map(JSONObject value) throws Exception {
                String sku_id = value.getString("sku_id");
                String user_id = value.getString("user_id");
                String order_id = value.getString("order_id");
                Double split_total_amount = value.getDouble("split_total_amount");
                HashSet<String> set = new HashSet<>();
                set.add(order_id);

                return TradeUserSpuOrderBean.builder()
                        .orderIdSet(set)
                        .skuId(sku_id)
                        .userId(user_id)
                        .orderAmount(split_total_amount)
                        .build();
            }
        });

        //异步IO  多线程     数据源
        SingleOutputStreamOperator<TradeUserSpuOrderBean> streamOperator = AsyncDataStream.unorderedWait(wTMap, new DimAsyncFunction<TradeUserSpuOrderBean>("DIM_SKU_INFO") {

            @Override
            public String getKey(TradeUserSpuOrderBean input) {
                return input.getSkuId();
            }

            @Override
            public void join(TradeUserSpuOrderBean input, JSONObject dimInfo) {

                input.setSpuId(dimInfo.getString("SPU_ID"));
                input.setTrademarkId(dimInfo.getString("TM_ID"));
                input.setCategory3Id(dimInfo.getString("CATEGORY3_ID"));
            }
        }, 100, TimeUnit.SECONDS);

        //TODO 9.分组、开窗、聚合
        KeyedStream<TradeUserSpuOrderBean, Tuple4<String, String, String, String>> keyedStream =
                streamOperator.keyBy(new KeySelector<TradeUserSpuOrderBean, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(TradeUserSpuOrderBean value) throws Exception {
                        return new Tuple4<>(
                                value.getUserId(),
                                value.getSpuId(),
                                value.getTrademarkId(),
                                value.getCategory3Id());
                    }
                });
        SingleOutputStreamOperator<TradeUserSpuOrderBean> reduceDS = keyedStream
                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
                .reduce(new ReduceFunction<TradeUserSpuOrderBean>() {
                    @Override
                    public TradeUserSpuOrderBean reduce(TradeUserSpuOrderBean value1, TradeUserSpuOrderBean value2) throws Exception {
                        value1.getOrderIdSet().addAll(value2.getOrderIdSet());

                        value1.setOrderAmount(value1.getOrderAmount() + value2.getOrderAmount());
                        return value1;
                    }
                }, new WindowFunction<TradeUserSpuOrderBean, TradeUserSpuOrderBean, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> key, TimeWindow window, Iterable<TradeUserSpuOrderBean> input, Collector<TradeUserSpuOrderBean> out) throws Exception {

                        TradeUserSpuOrderBean userSpuOrderBean = input.iterator().next();

                        userSpuOrderBean.setTs(System.currentTimeMillis());
                        userSpuOrderBean.setOrderCount((long) userSpuOrderBean.getOrderIdSet().size());
                        userSpuOrderBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        userSpuOrderBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));

                        out.collect(userSpuOrderBean);
                    }
                });
        //spuId    c3
        //TODO 10.关联spu,tm,category维表补充相应的信息
        //10.1 关联SPU表
        SingleOutputStreamOperator<TradeUserSpuOrderBean> reduceWithSpuDS = AsyncDataStream.unorderedWait(reduceDS,
                new DimAsyncFunction<TradeUserSpuOrderBean>("DIM_SPU_INFO") {
                    @Override
                    public String getKey(TradeUserSpuOrderBean input) {
                        return input.getSpuId();
                    }

                    @Override
                    public void join(TradeUserSpuOrderBean input, JSONObject dimInfo) {
                        input.setSpuName(dimInfo.getString("SPU_NAME"));
                    }
                }, 100, TimeUnit.SECONDS);

        //10.2 关联Tm表
        SingleOutputStreamOperator<TradeUserSpuOrderBean> reduceWithTmDS = AsyncDataStream.unorderedWait(reduceWithSpuDS,
                new DimAsyncFunction<TradeUserSpuOrderBean>("DIM_BASE_TRADEMARK") {
                    @Override
                    public String getKey(TradeUserSpuOrderBean input) {
                        return input.getTrademarkId();
                    }

                    @Override
                    public void join(TradeUserSpuOrderBean input, JSONObject dimInfo) {
                        input.setTrademarkName(dimInfo.getString("TM_NAME"));
                    }
                }, 100, TimeUnit.SECONDS);

        //10.3 关联Category3
        SingleOutputStreamOperator<TradeUserSpuOrderBean> reduceWithCategory3DS = AsyncDataStream.unorderedWait(reduceWithTmDS,
                new DimAsyncFunction<TradeUserSpuOrderBean>("DIM_BASE_CATEGORY3") {
                    @Override
                    public String getKey(TradeUserSpuOrderBean input) {
                        return input.getCategory3Id();
                    }

                    @Override
                    public void join(TradeUserSpuOrderBean input, JSONObject dimInfo) {
                        input.setCategory3Name(dimInfo.getString("NAME"));
                        input.setCategory2Id(dimInfo.getString("CATEGORY2_ID"));
                    }
                }, 100, TimeUnit.SECONDS);

        //10.4 关联Category2
        SingleOutputStreamOperator<TradeUserSpuOrderBean> reduceWithCategory2DS = AsyncDataStream.unorderedWait(reduceWithCategory3DS,
                new DimAsyncFunction<TradeUserSpuOrderBean>("DIM_BASE_CATEGORY2") {
                    @Override
                    public String getKey(TradeUserSpuOrderBean input) {
                        return input.getCategory2Id();
                    }

                    @Override
                    public void join(TradeUserSpuOrderBean input, JSONObject dimInfo) {
                        input.setCategory2Name(dimInfo.getString("NAME"));
                        input.setCategory1Id(dimInfo.getString("CATEGORY1_ID"));
                    }
                }, 100, TimeUnit.SECONDS);

        //10.5 关联Category1
        SingleOutputStreamOperator<TradeUserSpuOrderBean> reduceWithCategory1DS = AsyncDataStream.unorderedWait(reduceWithCategory2DS,
                new DimAsyncFunction<TradeUserSpuOrderBean>("DIM_BASE_CATEGORY1") {
                    @Override
                    public String getKey(TradeUserSpuOrderBean input) {
                        return input.getCategory1Id();
                    }

                    @Override
                    public void join(TradeUserSpuOrderBean input, JSONObject dimInfo) {
                        input.setCategory1Name(dimInfo.getString("NAME"));
                    }
                }, 100, TimeUnit.SECONDS);

        //TODO 11.将数据写出到ClickHouse
        reduceWithCategory1DS.print(">>>>>>>>>>>>>>>>>");

        env.execute();


    }

}